#!/bin/bash
# Confirm release() path and any existing sync/async mixing callers.
rg -nP -C3 '\bacquire_sync\b|registry\.release\b|ConcurrencyRegistry\b'

cat -n src/praisonai-agents/praisonaiagents/agent/concurrency.py

# Find actual usage of acquire_sync and release in test/agent code
rg -n "acquire_sync|\.release\(" src/praisonai-agents/praisonaiagents/ src/praisonai-agents/tests/ -A 2

# Check the test more carefully - is it in async context?
sed -n '101,110p' src/praisonai-agents/tests/unit/agent/test_agent_concurrency.py

# Also check if acquire_sync is called from actual async code anywhere
rg -n "acquire_sync" --type py

# Check Python version and asyncio.Semaphore behavior
python3 --version

# Let's create a test script to see what actually happens with cross-loop semaphore usage
cat > /tmp/test_semaphore.py << 'EOF'
import asyncio
import threading

# Test 1: Semaphore created, then acquire in separate loop
sem = asyncio.Semaphore(1)
print(f"1. Created sem: {sem}")
print(f"   sem._loop initially: {getattr(sem, '_loop', 'not set')}")

# Acquire in a new loop (the worker thread approach)
def acquire_in_new_loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    print(f"2. Worker thread loop: {loop}")
    try:
        loop.run_until_complete(sem.acquire())
        print(f"3. After acquire in worker, sem._loop: {getattr(sem, '_loop', 'not set')}")
    finally:
        loop.close()
    print(f"4. After loop.close() in worker")

t = threading.Thread(target=acquire_in_new_loop)
t.start()
t.join()

print(f"5. Main thread, after worker finishes")
print(f"   sem._loop now: {getattr(sem, '_loop', 'not set')}")

# Try to release from main thread (no loop running)
try:
    sem.release()
    print(f"6. Release succeeded from main thread")
except Exception as e:
    print(f"6. Release failed: {type(e).__name__}: {e}")

# Test 2: Create loop, acquire, close immediately
print("\n--- Test 2: Close loop immediately ---")
sem2 = asyncio.Semaphore(1)
loop2 = asyncio.new_event_loop()
asyncio.set_event_loop(loop2)
try:
    loop2.run_until_complete(sem2.acquire())
    print(f"Acquired in loop2, sem2._loop: {getattr(sem2, '_loop', 'not set')}")
finally:
    loop2.close()
    asyncio.set_event_loop(None)

print(f"After loop2.close(), sem2._loop: {getattr(sem2, '_loop', 'not set')}")
try:
    sem2.release()
    print(f"Release succeeded from main thread after loop close")
except Exception as e:
    print(f"Release failed: {type(e).__name__}: {e}")
EOF
python3 /tmp/test_semaphore.py

# Let's test the timeout behavior specifically
cat > /tmp/test_timeout.py << 'EOF'
import asyncio
import concurrent.futures
import time
import threading

# Create a semaphore that will never acquire (locked)
sem = asyncio.Semaphore(0)  # Start at 0, so acquire will block

def acquire_in_new_loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    print(f"[Worker] Starting acquire in new loop at {time.time():.2f}")
    try:
        # This will hang forever since sem is at 0
        loop.run_until_complete(sem.acquire())
        print(f"[Worker] Acquired (should not happen)")
    except asyncio.CancelledError:
        print(f"[Worker] Cancelled")
    finally:
        loop.close()
    print(f"[Worker] Finished at {time.time():.2f}")

start = time.time()
print(f"Starting at {start:.2f}")

try:
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
        print(f"Pool created at {time.time() - start:.2f}s")
        future = pool.submit(acquire_in_new_loop)
        try:
            result = future.result(timeout=2)
            print(f"Got result: {result}")
        except concurrent.futures.TimeoutError:
            print(f"[Main] result() raised TimeoutError at {time.time() - start:.2f}s")
            print(f"[Main] Now exiting 'with' block (this will block waiting for worker)...")
except Exception as e:
    print(f"[Main] Exception: {e}")

print(f"[Main] Exited 'with' block at {time.time() - start:.2f}s")
print(f"Total time elapsed: {time.time() - start:.2f}s")
EOF
timeout 10 python3 /tmp/test_timeout.py

# Test the timeout behavior without timeout command - use shorter timeout
cat > /tmp/test_timeout.py << 'EOF'
import asyncio
import concurrent.futures
import time
import threading

# Create a semaphore that will never acquire (locked)
sem = asyncio.Semaphore(0)  # Start at 0, so acquire will block

def acquire_in_new_loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    print(f"[Worker] Starting acquire in new loop at {time.time():.2f}")
    try:
        # This will hang forever since sem is at 0
        loop.run_until_complete(sem.acquire())
        print(f"[Worker] Acquired (should not happen)")
    except asyncio.CancelledError:
        print(f"[Worker] Cancelled")
    finally:
        loop.close()
    print(f"[Worker] Finished at {time.time():.2f}")

start = time.time()
print(f"Starting at {start:.2f}")

# Test without hanging forever - use very short timeout
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
    print(f"Pool created at {time.time() - start:.2f}s")
    future = pool.submit(acquire_in_new_loop)
    try:
        result = future.result(timeout=1)
        print(f"Got result: {result}")
    except concurrent.futures.TimeoutError:
        print(f"[Main] result(timeout=1) raised TimeoutError at {time.time() - start:.2f}s")
        print(f"[Main] Now about to exit 'with' block (ThreadPoolExecutor.__exit__ will do shutdown(wait=True))...")
        
print(f"[Main] Exited 'with' block at {time.time() - start:.2f}s")
print(f"Total time elapsed: {time.time() - start:.2f}s (should be >> 1 second due to shutdown wait)")
EOF

# Run with a shell timeout
( sleep 15; kill $$ 2>/dev/null ) & python3 /tmp/test_timeout.py ; pkill -P $$

# Check if Python 3.11 asyncio.Semaphore stores loop ref differently
python3 << 'EOF'
import asyncio
import sys

print(f"Python version: {sys.version}")

# Check Semaphore internals
sem = asyncio.Semaphore(1)
print(f"\nSemaphore attributes: {dir(sem)}")
print(f"\nSemaphore __dict__: {sem.__dict__}")

# After creating a semaphore, check what happens when we await it in different loops
import asyncio
loop1 = asyncio.new_event_loop()
asyncio.set_event_loop(loop1)

async def test1():
    print(f"\nIn loop1, sem.__dict__ before acquire: {sem.__dict__}")
    await sem.acquire()
    print(f"In loop1, sem.__dict__ after acquire: {sem.__dict__}")

loop1.run_until_complete(test1())

# Now try from a different loop
loop2 = asyncio.new_event_loop()
asyncio.set_event_loop(loop2)

async def test2():
    print(f"\nIn loop2, sem.__dict__ before acquire: {sem.__dict__}")
    await sem.acquire()
    print(f"In loop2, sem.__dict__ after acquire: {sem.__dict__}")

try:
    loop2.run_until_complete(test2())
except RuntimeError as e:
    print(f"In loop2, got error: {e}")

loop1.close()
loop2.close()
EOF

# Simpler check - just look at asyncio.Semaphore source behavior without executing problematic code
python3 -c "
import asyncio
import inspect

# Get the source of asyncio.Semaphore to understand how it handles loops
print('=== asyncio.Semaphore.acquire source ===')
print(inspect.getsource(asyncio.Semaphore.acquire))
"

python3 -c "
import asyncio
import inspect

# Get the source of asyncio.Semaphore._get_loop to understand loop binding
print('=== asyncio.Semaphore._get_loop source ===')
try:
    print(inspect.getsource(asyncio.Semaphore._get_loop))
except:
    print('_get_loop not found in Semaphore, checking base Lock...')
    print(inspect.getsource(asyncio.Lock._get_loop))
"

python3 -c "
import asyncio
import inspect

# Get the source of asyncio.Semaphore.release to understand how it handles the loop
print('=== asyncio.Semaphore.release source ===')
try:
    print(inspect.getsource(asyncio.Semaphore.release))
except:
    print('Checking base Lock.release...')
    print(inspect.getsource(asyncio.Lock.release))
"

# Now test the actual critical issue: RuntimeError when semaphore is bound to wrong loop
python3 << 'EOF'
import asyncio
import concurrent.futures

sem = asyncio.Semaphore(1)
print("Initial state: sem._loop =", sem._loop)

# Simulate the problematic path: acquire in worker thread
def acquire_in_worker():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    print("Worker: before acquire, sem._loop =", sem._loop)
    loop.run_until_complete(sem.acquire())
    print("Worker: after acquire, sem._loop =", sem._loop)
    loop.close()
    print("Worker: loop closed")

# Run it
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
    pool.submit(acquire_in_worker).result()

print("Main: after worker finished, sem._loop =", sem._loop)

# Now try to use semaphore from async context
async def use_semaphore():
    print("Async: attempting acquire with sem._loop =", sem._loop)
    try:
        await sem.acquire()
        print("Async: acquire succeeded")
    except RuntimeError as e:
        print(f"Async: RuntimeError: {e}")

# Create a new loop in the main thread
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
    loop.run_until_complete(use_semaphore())
finally:
    loop.close()
EOF

# Minimal test without actual event loop execution
python3 << 'EOF'
import asyncio

# Test what _get_loop does when called from non-running context
sem = asyncio.Semaphore(1)
print("Semaphore._loop initially:", sem._loop)

# Check if calling _get_loop outside running loop works
try:
    result = sem._get_loop()
    print("_get_loop() succeeded:", result)
except RuntimeError as e:
    print("_get_loop() raises RuntimeError:", e)
EOF

python3 << 'EOF'
import asyncio

# Check _wake_up_next implementation
import inspect
print("=== Semaphore._wake_up_next source ===")
try:
    print(inspect.getsource(asyncio.Semaphore._wake_up_next))
except AttributeError:
    # Try base class
    print(inspect.getsource(asyncio.Lock._wake_up_next))
EOF

# Check if the test actually runs the problematic code path
# The test_sync_acquire_release is NOT in an async test, so it won't hit the RuntimeError branch
sed -n '101,115p' src/praisonai-agents/tests/unit/agent/test_agent_concurrency.py

# Verify: is it decorated with `@pytest.mark.asyncio`?
sed -n '100,115p' src/praisonai-agents/tests/unit/agent/test_agent_concurrency.py

# Check if any test calls acquire_sync from within an `@pytest.mark.asyncio` test
rg -B 5 "acquire_sync" src/praisonai-agents/tests/unit/agent/test_agent_concurrency.py

# Search for actual usage of acquire_sync in agent code (not tests)
rg -n "acquire_sync" src/praisonai-agents/praisonaiagents/ --type py --not-tests

# Also check if there are any agent implementations that mix sync and async
rg -B5 -A5 "def.*acquire" src/praisonai-agents/praisonaiagents/agent/ --type py | head -50